/*******************************************************************************
 *
 * Pentaho Data Integration
 *
 * Copyright (C) 2002-2012 by Pentaho : http://www.pentaho.com
 *
 *******************************************************************************
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with
 * the License. You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 ******************************************************************************/

package org.pentaho.di.trans.steps.esinput;


import org.pentaho.di.core.Const;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.row.RowDataUtil;
import org.pentaho.di.core.row.RowMeta;
import org.pentaho.di.core.row.ValueMeta;
import org.pentaho.di.core.row.ValueMetaInterface;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.BaseStep;
import org.pentaho.di.trans.step.StepDataInterface;
import org.pentaho.di.trans.step.StepInterface;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.step.StepMetaInterface;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;

public class ESInput extends BaseStep implements StepInterface {

	private ESInputMeta meta;
	private ESInputData data;

	private String host;
	private int port;
	private String index;
	private String type;
	private Boolean _id;
	private String[] target;
	private String[] targetType;
//	private JestClient client;
	private StringBuilder result;
	private List<Map<String, Object>> list;
	Object[] values;
	ValueMetaInterface valueMeta[];

	private static String scrollId = null;

	public ESInput(StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
			Trans trans) {
		super(stepMeta, stepDataInterface, copyNr, transMeta, trans);
		list = new ArrayList<Map<String, Object>>();
	}

	/**
	 * 按行处理输入的数据流 按照提交数量批量提交数据到ES
	 * 
	 * @throws IOException
	 */
	@Override
	public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException {
		meta = (ESInputMeta) smi;
		data = (ESInputData) sdi;

		if (first) {
			first = false;
			host = meta.getEsHost();
			port = meta.getEsPort();
			index = meta.getEsIndex();
			type = meta.getEsType();
			_id = meta.isAutoID();
			target = meta.getFieldDatabase();
			target= Arrays.copyOf(target, target.length+1);
			target[target.length-1]="ESType";

			targetType = meta.getFieldDatabaseType();
			targetType= Arrays.copyOf(targetType, targetType.length+1);
			targetType[targetType.length-1]="String";

//			client = ESUtils.getJestClient(host, port+"");

			//				result = new StringBuilder(startScrollSearch(client, index, type, 10000));
//				list = ESUtils.parse(result.toString(), _id);
			if (list.size() == 0) {
				setOutputDone();
				return false;
			}

			data.outputRowMeta = new RowMeta();
			valueMeta = new ValueMeta[target.length];
			for (int len = 0; len < valueMeta.length; len++) {
				// 构造一个新的输出列。
				if("Date".equalsIgnoreCase(targetType[len])||"BigNumber".equalsIgnoreCase(targetType[len])){
					targetType[len] = "String";
				}
				int valtype = ValueMeta.getType(targetType[len]);
				valueMeta[len] = new ValueMeta(target[len], valtype);
				valueMeta[len].setLength(-1);
				data.outputRowMeta.addValueMeta(valueMeta[len]);
			}


			for (int i = 0; i < list.size(); i++) {
				values = new Object[data.outputRowMeta.size()];
				for (int len = 0; len < data.outputRowMeta.size(); len++) {
					Object tmp = list.get(i).get(target[len]);
					if(tmp==null){
						values[len] = type;
						continue;
					}
					switch (tmp.getClass().getSimpleName()) {
					case "Integer":
						values[len] = new Long(tmp.toString());
						break;
					case "Double":
						values[len] = new Double(tmp.toString());
						break;
					case "Float":
						values[len] = new Double(tmp.toString());
						break;
					case "Boolean":
						values[len] = new Boolean(tmp.toString());
						break;
					default:
						if (valueMeta[len].isString()) {
							values[len] = tmp.toString();
							break;
						} else if (valueMeta[len].isDate()) {
							//data.outputRowMeta.removeValueMeta(len);
							//data.outputRowMeta.addValueMeta(len, meta);;
							//valueMeta[len].toStringMeta();
								values[len] = tmp.toString();
							break;
						}
					}
				}

				putRow(data.outputRowMeta, values);
			}
			return true;
		}


		//			result = new StringBuilder(readMoreFromSearch(client, scrollId, 10000));
//			list.clear();
//			list = ESUtils.parse(result.toString(), _id);
		if (list.size() == 0) {
			setOutputDone();
			System.out.println("list.size()==0");
			return false;
		}
		for (int i = 0; i < list.size(); i++) {
			values = RowDataUtil.allocateRowData(data.outputRowMeta.size());
			for (int len = 0; len < target.length; len++) {
				Object tmp = list.get(i).get(target[len]);
				if(tmp==null){
					values[len] = type;
					continue;
				}
				switch (tmp.getClass().getSimpleName()) {
				case "Integer":
					values[len] = new Long(tmp.toString());
					break;
				case "Double":
					values[len] = new Double(tmp.toString());
					break;
				case "Float":
					values[len] = new Double(tmp.toString());
					break;
				case "Boolean":
					values[len] = new Boolean(tmp.toString());
					break;
				default:
					if (valueMeta[len].isString()) {
						values[len] = tmp.toString();
						break;
					} else if (valueMeta[len].isDate()) {
						values[len] = tmp.toString();
						break;
					}
				}
			}
			putRow(data.outputRowMeta, values);
		}
		return true;
	}

	public boolean init(StepMetaInterface smi, StepDataInterface sdi) {
//		meta = (esInput.ESInputMeta) smi;
		data = (ESInputData) sdi;
		return super.init(smi, sdi);
	}

	public void dispose(StepMetaInterface smi, StepDataInterface sdi) {
//		meta = (esInput.ESInputMeta) smi;
//		data = (ESInputData) sdi;
//		client.shutdownClient();
		super.dispose(smi, sdi);
	}

	public void run() {
		logBasic("Starting to run...");
		try {// 运行processRow方法
			while (processRow(meta, data) && !isStopped())
				;
//			client.shutdownClient();
		} catch (final Exception e) {
			logError("Unexpected error : " + e.toString());
			logError(Const.getStackTracker(e));
			setErrors(1);
			stopAll();
		} finally {
			dispose(meta, data);
			logBasic("Finished, processing " + linesRead + " rows");
			markStop();
		}
	}

	public int getFieldIndex(final String[] fields, final String fValue) {
		int i;
		for (i = 0; (i < fields.length) && (!(fields[i].toUpperCase().equals(fValue.toUpperCase()))); i++)
			;
		return (i < fields.length) ? i : -1;
	}

	/*

	public static String readMoreFromSearch(JestClient client, String scrollId, int size) throws IOException {
		SearchScroll scroll = new SearchScroll.Builder(scrollId, "1m").setParameter(Parameters.SIZE, size).build();
		JestResult searchResult = client.execute(scroll);
		return searchResult.getJsonString();
	}

	public static String startScrollSearch(JestClient client, String index, String type, int size) throws IOException {
		SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
		searchSourceBuilder.query(QueryBuilders.matchAllQuery());
		Search search = new Search.Builder(searchSourceBuilder.toString()).addIndex(index).addType(type)
				.setParameter(Parameters.SIZE, size).setParameter(Parameters.SCROLL, "1m").build();
		SearchResult searchResult = client.execute(search);
		// scrollId应为全局变量
		scrollId = searchResult.getJsonObject().get("_scroll_id").getAsString();
		return searchResult.getJsonString();
	}
	*/

	
}